from pyflink.datastream import StreamExecutionEnvironment, KeySelector

env = StreamExecutionEnvironment.get_execution_environment()

env.set_parallelism(1)

studnets_ds = env.read_text_file('../../data/students.txt')

#1、lambda
key_by_ds = studnets_ds.map(lambda line: (line.split(',')[-1],1))
## key_by: 将相同的key发送到下游同一个task中
# key_by_ds.key_by(lambda kv: kv[0]).sum(1).print()


## 使用类的方式

class MyKeyClass(KeySelector):

    def get_key(self, value):
        return value[0]

key_by_ds.key_by(MyKeyClass()).sum(1).print()

env.execute()